package com.atom.zx.concurrency;

import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.TimeUnit;

import com.atom.zx.Scheduler;
import com.atom.zx.Subscription;
import com.atom.zx.subscriptions.Subscriptions;
import com.atom.zx.util.Logs;
import com.atom.zx.util.functions.Func2;

public class TestScheduler extends Scheduler {
    private final Queue<TimedAction<?>> queue = new PriorityQueue<TimedAction<?>>(11, new CompareActionsByTime());

    private static class TimedAction<T> {
        private final long time;
        private final Func2<Scheduler, T, Subscription> action;
        private final T state;
        private final TestScheduler scheduler;

        private TimedAction(TestScheduler scheduler, long time, Func2<Scheduler, T, Subscription> action, T state) {
            this.time = time;
            this.action = action;
            this.state = state;
            this.scheduler = scheduler;
        }

        @Override
        public String toString() {
            return String.format("TimedAction(time = %d, action = %s)", time, action.toString());
        }
    }

    private static class CompareActionsByTime implements Comparator<TimedAction<?>> {
        @Override
        public int compare(TimedAction<?> action1, TimedAction<?> action2) {
            return Long.valueOf(action1.time).compareTo(Long.valueOf(action2.time));
        }
    }

    private long time;

    @Override
    public long now() {
        return TimeUnit.NANOSECONDS.toMillis(time);
    }

    public void advanceTimeBy(long delayTime, TimeUnit unit) {
        Logs.e("advanceTimeBy = " + (time + unit.toNanos(delayTime)) , "currTime = "+time);
        advanceTimeTo(time + unit.toNanos(delayTime), TimeUnit.NANOSECONDS);
    }

    public void advanceTimeTo(long delayTime, TimeUnit unit) {
        long targetTime = unit.toNanos(delayTime);
        Logs.e("advanceTimeTo = " + targetTime);
        triggerActions(targetTime);
    }

    public void triggerActions() {
        triggerActions(time);
    }

    @SuppressWarnings("unchecked")
    private void triggerActions(long targetTimeInNanos) {
        while (!queue.isEmpty()) {
            TimedAction<?> current = queue.peek();
            if (current.time > targetTimeInNanos) {
                break;
            }
            time = current.time;
            queue.remove();
            // because the queue can have wildcards we have to ignore the type T for the state
            ((Func2<Scheduler, Object, Subscription>) current.action).call(current.scheduler, current.state);
        }
        time = targetTimeInNanos;
    }

    @Override
    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action) {
        return schedule(state, action, 0, TimeUnit.MILLISECONDS);
    }

    @Override
    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit) {
        queue.add(new TimedAction<T>(this, time + unit.toNanos(delayTime), action, state));
        return Subscriptions.empty();
    }
}
